package com.ams.protocol.rtmp.net;

import java.io.EOFException;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ams.protocol.rtmp.amf.*;
import com.ams.io.RandomAccessFileWriter;
import com.ams.media.flv.FlvException;
import com.ams.media.flv.FlvSerializer;
import com.ams.protocol.rtmp.message.*;
import com.ams.protocol.rtmp.*;

public class NetConnection {
    final private Logger logger = LoggerFactory.getLogger(NetConnection.class);

    private RtmpHandShake handshake;
    private RtmpConnection rtmp;
    private NetContext context;
    private NetStreamManager streamManager;
    private HashMap<Integer, StreamPlayer> streamPlayers;
    private HashMap<Integer, StreamPublisher> streamPublishers;

    public NetConnection(RtmpConnection rtmp, NetContext context) {
        this.rtmp = rtmp;
        this.handshake = new RtmpHandShake(rtmp);
        this.streamManager = new NetStreamManager();
        this.streamPlayers = new HashMap<Integer, StreamPlayer>();
        this.streamPublishers = new HashMap<Integer, StreamPublisher>();
        this.context = context;
    }

    private void onMediaMessage(RtmpHeader header, RtmpMessage message)
            throws NetConnectionException, IOException {
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            throw new NetConnectionException("Unknown stream "
                    + header.getStreamId());
        }

        StreamPublisher publisher = streamPublishers.get(header.getStreamId());
        if (publisher == null) {
            return;
        }
        publisher.publish(message.toMediaMessage(header.getTimestamp()));
        if (publisher.isPing()) {
            rtmp.writeProtocolControlMessage(new RtmpMessageAck(publisher
                    .getPingBytes()));
        }
        
    }

    private void onSharedMessage(RtmpHeader header, RtmpMessage message) {
        // TODO
    }

    private void onCommandMessage(RtmpHeader header, RtmpMessage message)
            throws NetConnectionException, IOException, FlvException {
        RtmpMessageCommand command = (RtmpMessageCommand) message;
        String cmdName = command.getName();
        logger.info("command: {}", cmdName);
        if ("connect".equals(cmdName)) {
            onConnect(header, command);
        } else if ("createStream".equals(cmdName)) {
            onCreateStream(header, command);
        } else if ("deleteStream".equals(cmdName)) {
            onDeleteStream(header, command);
        } else if ("closeStream".equals(cmdName)) {
            onCloseStream(header, command);
        } else if ("getStreamLength".equals(cmdName)) {
            onGetStreamLength(header, command);
        } else if ("play".equals(cmdName)) {
            onPlay(header, command);
        } else if ("play2".equals(cmdName)) {
            onPlay2(header, command);
        } else if ("publish".equals(cmdName)) {
            onPublish(header, command);
        } else if ("pause".equals(cmdName) || "pauseRaw".equals(cmdName)) {
            onPause(header, command);
        } else if ("receiveAudio".equals(cmdName)) {
            onReceiveAudio(header, command);
        } else if ("receiveVideo".equals(cmdName)) {
            onReceiveVideo(header, command);
        } else if ("seek".equals(cmdName)) {
            onSeek(header, command);
        } else { // remote rpc call
            onCall(header, command);
        }
    }

    private void onConnect(RtmpHeader header, RtmpMessageCommand command)
            throws NetConnectionException, IOException {
        AmfValue amfObject = command.getCommandObject();
        Map<String, AmfValue> obj = amfObject.object();

        String app = obj.get("app").string();
        if (app == null) {
            netConnectionError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "Invalid 'Connect' parameters");
            return;
        }
        logger.info("connected app: {}", app);
        
        context.setAttribute("app", app);
        rtmp.writeProtocolControlMessage(new RtmpMessageWindowAckSize(
                128 * 1024));
        rtmp.writeProtocolControlMessage(new RtmpMessagePeerBandwidth(
                128 * 1024, (byte) 2));
        rtmp.writeProtocolControlMessage(new RtmpMessageUserControl(
                RtmpMessageUserControl.EVT_STREAM_BEGIN, header.getStreamId()));

        AmfValue value0 = AmfValue.newObject();
        value0.put("capabilities", 31).put("fmsver", "AMS/0,1,0,0")
                .put("mode", 1);

        AmfValue value = AmfValue.newObject();
        value.put("level", "status")
                .put("code", "NetConnection.Connect.Success")
                .put("description", "Connection succeeded.");

        AmfValue objectEncoding = obj.get("objectEncoding");
        if (objectEncoding != null) {
            value.put("objectEncoding", objectEncoding);
        }

        rtmp.writeRtmpMessage(header.getChunkStreamId(), 0, 0,
                new RtmpMessageCommand("_result", command.getTransactionId(),
                        AmfValue.array(value0, value)));

    }

    private void onGetStreamLength(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException, FlvException {
        AmfValue[] args = command.getArgs();
        String streamName = args[1].string();
        // rtmp.writeRtmpMessage(header.getChunkStreamId(), 0, 0,
        // new RtmpMessageCommand("_result", command.getTransactionId(),
        // AmfValue.array(null, 140361)));
    }

    private void onPlay(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException, FlvException {
        AmfValue[] args = command.getArgs();
        String streamName = args[1].string();
        int start = (args.length > 2) ? args[2].integer() : -2;
        int duration = (args.length > 3) ? args[3].integer() : -1;
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "Invalid 'Play' stream id "
                            + header.getStreamId());
            return;
        }
        stream.setTransactionId(command.getTransactionId());
        stream.setStreamName(streamName);

        logger.info("play stream: {}", streamName);

        // write status message
        stream.writePlayMessage();

        // create player
        String app = context.getAttribute("app");
        switch (start) {
        case -1: // live only
        {
            String publishName = context.getPublishName(app, streamName);
            StreamPublisher publisher = StreamPublisherManager
                    .getPublisher(publishName);
            if (publisher == null) {
                stream.writeErrorMessage("Unknown stream '" + streamName
                        + "'");
                return;
            }
            publisher.addSubscriber(stream);
        }
            break;
        case -2: // first find live
        {
            String publishName = context.getPublishName(app, streamName);
            StreamPublisher publisher = StreamPublisherManager
                    .getPublisher(publishName);
            if (publisher != null) {
                publisher.addSubscriber(stream);
            } else {
                String tokens[] = streamName.split(":");
                String type = "";
                String file = streamName;
                if (tokens.length >= 2) {
                    type = tokens[0];
                    file = tokens[1];
                }
                String path = context.getRealPath(app, file, type);
                StreamPlayer player = StreamPlayer.createPlayer(stream, type,
                        path);
                if (player != null) {
                    streamPlayers.put(header.getStreamId(), player);
                    player.seek(0);
                }
            }
        }
            break;
        default: // >= 0
            String tokens[] = streamName.split(":");
            String type = "";
            String file = streamName;
            if (tokens.length >= 2) {
                type = tokens[0];
                file = tokens[1];
            }
            String path = context.getRealPath(app, file, type);
            StreamPlayer player = StreamPlayer.createPlayer(stream, type, path);
            if (player != null) {
                streamPlayers.put(header.getStreamId(), player);
                player.seek(start);
            }
        }
    }

    private void onPlay2(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException, FlvException {
        AmfValue[] args = command.getArgs();
        int startTime = args[0].integer();
        String oldStreamName = args[1].string();
        String streamName = args[2].string();
        int duration = (args.length > 3) ? args[3].integer() : -1;
        String transition = (args.length > 4) ? args[4].string() : "switch"; // switch
                                                                             // or
                                                                             // swap

        // TODO
    }

    private void onSeek(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException, FlvException {
        AmfValue[] args = command.getArgs();
        int offset = args[1].integer();
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "Invalid 'Seek' stream id "
                            + header.getStreamId());
            return;
        }
        stream.setTransactionId(command.getTransactionId());
        int streamId = header.getStreamId();
        StreamPlayer player = streamPlayers.get(streamId);
        if (player == null) {
            stream.writeErrorMessage("Invalid 'Seek' stream id " + streamId);
            return;
        }

        stream.writeSeekMessage(offset);
    }

    private void onPause(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException, FlvException {
        AmfValue[] args = command.getArgs();
        boolean pause = args[1].bool();
        int time = args[2].integer();
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "Invalid 'Pause' stream id "
                            + header.getStreamId());
            return;
        }
        stream.setTransactionId(command.getTransactionId());
        StreamPlayer player = streamPlayers.get(header.getStreamId());
        if (player == null) {
            stream.writeErrorMessage("This channel is already closed");
            return;
        }
        logger.info("pause stream: {}", stream.getStreamName());

        player.pause(pause);

        stream.writePauseMessage(pause);
    }

    private void onPublish(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException {
        AmfValue[] args = command.getArgs();
        String publishName = args[1].string();
        if (StreamPublisherManager.getPublisher(publishName) != null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "The publish '" + publishName
                            + "' is already used");
            return;
        }

        String type = (args.length > 2) ? args[2].string() : null;
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(), "Invalid 'Publish' stream id "
                            + header.getStreamId());
            return;
        }
        stream.setTransactionId(command.getTransactionId());

        String app = context.getAttribute("app");
        String streamName = publishName.split("\\?")[0];
        logger.info("publish stream: {}", streamName);

        // save to share or file
        String publishFullName = context.getPublishName(app, streamName);
        StreamPublisher publisher = new StreamPublisher(publishFullName);
        streamPublishers.put(header.getStreamId(), publisher);
        if ("record".equals(type)) {
            String file = context.getRealPath(app, streamName, "");
            try {
                RandomAccessFileWriter writer = new RandomAccessFileWriter(
                        file, false);
                publisher.setRecorder(new FlvSerializer(writer));
            } catch (IOException e) {
                logger.debug(e.getMessage());
            }
        } else if ("append".equals(type)) {
            String file = context.getRealPath(app, streamName, "");
            RandomAccessFileWriter writer = new RandomAccessFileWriter(file,
                    true);
            publisher.setRecorder(new FlvSerializer(writer));
        } else if ("live".equals(type)) {
            // nothing to do
        }
        StreamPublisherManager.addPublisher(publisher);

        stream.writePublishMessage(publishName);
    }

    private void onReceiveAudio(RtmpHeader header, RtmpMessageCommand command)
            throws IOException {
        AmfValue[] args = command.getArgs();
        boolean flag = args[1].bool();
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(),
                    "Invalid 'ReceiveAudio' stream id " + header.getStreamId());
            return;
        }
        StreamPlayer player = streamPlayers.get(header.getStreamId());
        if (player != null) {
            player.audioPlaying(flag);
        }
    }

    private void onReceiveVideo(RtmpHeader header, RtmpMessageCommand command)
            throws IOException {
        AmfValue[] args = command.getArgs();
        boolean flag = args[1].bool();
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(),
                    "Invalid 'ReceiveVideo' stream id " + header.getStreamId());
            return;
        }
        StreamPlayer player = streamPlayers.get(header.getStreamId());
        if (player != null) {
            player.videoPlaying(flag);
        }
    }

    private void onCreateStream(RtmpHeader header, RtmpMessageCommand command)
            throws IOException {
        NetStream stream = streamManager.createStream(rtmp);
        rtmp.writeRtmpMessage(header.getChunkStreamId(), 0, 0,
                new RtmpMessageCommand("_result", command.getTransactionId(),
                        AmfValue.array(null, stream.getStreamId())));

    }

    private void onDeleteStream(RtmpHeader header, RtmpMessageCommand command)
            throws IOException, NetConnectionException {
        AmfValue[] args = command.getArgs();
        int streamId = args[1].integer();
        NetStream stream = streamManager.get(streamId);
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(),
                    "Invalid 'deleteStream' stream id");
            return;
        }
        streamManager.removeStream(stream);
    }

    private void onCloseStream(RtmpHeader header, RtmpMessageCommand command)
            throws NetConnectionException, IOException {
        NetStream stream = streamManager.get(header.getStreamId());
        if (stream == null) {
            streamError(header.getChunkStreamId(), header.getStreamId(),
                    command.getTransactionId(),
                    "Invalid 'CloseStream' stream id " + header.getStreamId());
            return;
        }
        logger.info("close stream: {}", stream.getStreamName());
        streamManager.removeStream(stream);
    }

    private void onCall(RtmpHeader header, RtmpMessageCommand command) {
        // String procedureName = command.getName();
    }

    public void writeError(int chunkStreamId, int streamId, int transactionId,
            String code, String msg) throws IOException {
        AmfValue value = AmfValue.newObject();
        value.put("level", "error").put("code", code).put("details", msg);
        rtmp.writeRtmpMessage(
                chunkStreamId,
                streamId,
                0,
                new RtmpMessageCommand("onStatus", transactionId, AmfValue
                        .array(null, value)));
    }

    public void netConnectionError(int chunkStreamId, int streamId,
            int transactionId, String msg) throws IOException {
        writeError(chunkStreamId, streamId, transactionId, "NetConnection", msg);
    }

    public void streamError(int chunkStreamId, int streamId, int transactionId,
            String msg) throws IOException {
        writeError(chunkStreamId, streamId, transactionId, "NetStream.Error",
                msg);
    }

    public void readAndProcessRtmpMessage() throws IOException, RtmpException,
            AmfException {
        if (!handshake.isHandshakeDone()) {
            handshake.doServerHandshake();
            return;
        }
        
        do {
            if (!rtmp.readRtmpMessage()) {
                return;
            }
        } while (!rtmp.isRtmpMessageReady());
        
        RtmpHeader header = rtmp.getCurrentHeader();
        RtmpMessage message = rtmp.getCurrentMessage();
        try {
            switch (message.getType()) {
            case RtmpMessage.MESSAGE_AMF0_COMMAND:
            case RtmpMessage.MESSAGE_AMF3_COMMAND:
                onCommandMessage(header, message);
                break;
            case RtmpMessage.MESSAGE_AUDIO:
            case RtmpMessage.MESSAGE_VIDEO:
            case RtmpMessage.MESSAGE_AMF0_DATA:
            case RtmpMessage.MESSAGE_AMF3_DATA:
                onMediaMessage(header, message);
                break;
            case RtmpMessage.MESSAGE_USER_CONTROL:
                RtmpMessageUserControl userControl = (RtmpMessageUserControl) message;
                logger.debug(
                        "read message USER_CONTROL: {}, {}, {}",
                        new Integer[] { userControl.getEvent(),
                                userControl.getStreamId(),
                                userControl.getTimestamp() });
                break;
            case RtmpMessage.MESSAGE_SHARED_OBJECT:
                onSharedMessage(header, message);
                break;
            case RtmpMessage.MESSAGE_CHUNK_SIZE:
                RtmpMessageChunkSize chunkSize = (RtmpMessageChunkSize) message;
                logger.debug("read message chunk size: {}",
                        chunkSize.getChunkSize());
                break;
            case RtmpMessage.MESSAGE_ABORT:
                RtmpMessageAbort abort = (RtmpMessageAbort) message;
                logger.debug("read message abort: {}", abort.getStreamId());
                break;
            case RtmpMessage.MESSAGE_ACK:
                RtmpMessageAck ack = (RtmpMessageAck) message;
                // logger.debug("read message ack: {}", ack.getBytes());
                break;
            case RtmpMessage.MESSAGE_WINDOW_ACK_SIZE:
                RtmpMessageWindowAckSize ackSize = (RtmpMessageWindowAckSize) message;
                logger.debug("read message window ack size: {}",
                        ackSize.getSize());
                break;
            case RtmpMessage.MESSAGE_PEER_BANDWIDTH:
                RtmpMessagePeerBandwidth peer = (RtmpMessagePeerBandwidth) message;
                logger.debug("read message peer bandwidth: {}, {}",
                        peer.getWindowAckSize(), peer.getLimitType());
                break;
            // case RtmpMessage.MESSAGE_AGGREGATE:
            // break;
            case RtmpMessage.MESSAGE_UNKNOWN:
                logger.warn("read message UNKNOWN!");
                break;
            }
        } catch (Exception e) {
            logger.debug("Exception", e);
        }
    }

    public void play() throws IOException {
        for (StreamPlayer player : streamPlayers.values()) {
            try {
                player.play();
            } catch (EOFException e) {
                player.pause(true);
                NetStream stream = player.getStream();
                stream.writeStopMessage();
            }
        }
        rtmp.flush();
    }

}
